package com.whz.disruptor.broker;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
import com.whz.disruptor.event.MessageEvent;
import com.whz.disruptor.handler.ConsumerHandler;
import com.whz.disruptor.properties.DisruptorProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

/**
 * 队列broker,处理消费和生产
 *
 * @author whz
 */
@Slf4j
public class DisruptorBrokerBean implements DisposableBean, InitializingBean {
    private final DisruptorProperties disruptorProperties;
    private final ConsumerHandler[] consumerHandler;
    private Disruptor<MessageEvent> disruptor;

    public DisruptorBrokerBean(DisruptorProperties disruptorProperties, ConsumerHandler... consumerHandler) {
        this.disruptorProperties = disruptorProperties;
        this.consumerHandler = consumerHandler;
    }

    public RingBuffer<MessageEvent> getRingBuffer() {
        return disruptor.getRingBuffer();
    }

    private void start() {
        disruptor = new Disruptor<>(
                MessageEvent::new,
                disruptorProperties.getRingBufferSize(),
                DaemonThreadFactory.INSTANCE,
                disruptorProperties.getProducerType(),
                disruptorProperties.getWaitStrategy());
        disruptor.handleEventsWith(consumerHandler);
        disruptor.start();
    }

    @Override
    public void afterPropertiesSet() {
        start();
    }

    @Override
    public void destroy() {
        if (disruptor != null) {
            disruptor.shutdown();
        }
    }
}